#include "EventLoop.h"
#include "TcpConnection.h"
#include "Acceptor.h"

#include <assert.h>
#include <unistd.h>
#include <sys/eventfd.h>
#include <iostream>

using namespace std;
namespace xkk
{

EventLoop::EventLoop(Acceptor &acceptor)
    :_efd(createEpollfd())
     ,_eventfd(createEventfd())
     ,_acceptor(acceptor)
     ,_evtList(1024)
     ,_isLooping(false)
    {
        addEpollReadFd(_acceptor.fd());
        addEpollReadFd(_eventfd);
    }

void EventLoop::loop()
{
    if(!_isLooping){
        _isLooping=true;
    }
    while(_isLooping){
        waitEpollFd();
    }
}

void EventLoop::unloop()
{
    if(_isLooping)
        _isLooping=false;
}

void EventLoop::runInLoop(Functor && cb)
{
    {
        MutexLockGuard autolock(_mutex);
        _pendingFunctors.push_back(std::move(cb));
    }
    wakeup();
}

void EventLoop::waitEpollFd()
{
    int nready=0;
    do{
        nready=::epoll_wait(_efd,&*_evtList.begin(),
                            _evtList.size(),5000);
    }while(nready==-1&&errno==EINTR);

    if(nready==-1){
        perror("epoll_wait");
        return;
    }
    else if(nready==0){
        printf(">>epoll_wait timeout!\n");
    }
    else{
        if(nready==_evtList.size()){
            _evtList.resize(2*nready);
        }
        for(auto idx=0;idx!=nready;idx++)
        {
            int fd=_evtList[idx].data.fd;
            if(fd==_acceptor.fd()){
                if(_evtList[idx].events&EPOLLIN){
                    handleNewConnection();
                }

            }else if(fd==_eventfd){
                if(_evtList[idx].events &EPOLLIN){
                    handleRead();
                    /* cout<<"doPendingFunctors()"<<endl; */
                    doPendingFunctors();
                 }
            }else{
                if(_evtList[idx].events &EPOLLIN){
                    handleMessage(fd);
                }
            }
        }
    }

}

void EventLoop::handleNewConnection()
{
    int peerfd=_acceptor.accept();
    addEpollReadFd(peerfd);
    TcpConnectionPtr conn(new TcpConnection(peerfd,this));
    conn->setConnectionCallback(_onConnection);
    conn->setMessageCallback(_onMessage);
    conn->setCloseCallback(_onClose);

    _conns.insert(make_pair(peerfd,conn));

    conn->handleConnectionCallback();
}

void EventLoop::handleMessage(int fd)
{
    auto iter=_conns.find(fd);
    assert(iter!=_conns.end());

    bool closed=isConnectionClosed(fd);
    if(closed){
        delEpollReadFd(fd);
        iter->second->handleCloseCallback();
        _conns.erase(iter);
    }else{
        iter->second->handleMessageCallback();
    }
}

bool EventLoop::isConnectionClosed(int fd)
{
    int ret;
    char buff[128]={0};
    do{
        ret=::recv(fd,buff,sizeof(buff),MSG_PEEK);
    }while(ret==-1&&errno==EINTR);
    return ret==0;
}

int EventLoop::createEpollfd()
{
    int fd=::epoll_create1(0);
    if(fd<0){
        perror("epoll_create1");
    }
    return fd;
}

void EventLoop::addEpollReadFd(int fd)
{
    struct epoll_event value;
    value.data.fd=fd;
    value.events=EPOLLIN;
    int ret=::epoll_ctl(_efd,EPOLL_CTL_ADD,fd,&value);
    if(-1==ret){
        perror("epoll_ctl");
    }
}

void EventLoop::delEpollReadFd(int fd)
{
    struct epoll_event value;
    value.data.fd=fd;
    int ret=::epoll_ctl(_efd,EPOLL_CTL_DEL,fd,&value);
    if(-1==ret){
        perror("epoll_ctl");
    }
}

int EventLoop::createEventfd()
{
    int fd=::eventfd(0,0);
    if(fd<0){
        perror("eventfd");
    }
    return fd;
}

void EventLoop::handleRead()
{
    uint64_t howmany;
    int ret=::read(_eventfd,&howmany,sizeof(howmany));
    if(ret!=sizeof(howmany)){
        perror("read");
    }
}

void EventLoop::wakeup()
{
    uint64_t one=1;
    int ret=::write(_eventfd,&one,sizeof(one));
    if(ret!=sizeof(one)){
        perror("write");
    }
}

void EventLoop::doPendingFunctors()
{
    vector<Functor> tmp;
    {
        MutexLockGuard autolock(_mutex);
        _pendingFunctors.swap(tmp);
    }
    for(auto &functor:tmp)
        functor();
}
}//end of namespace xkk
